package com.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.app.function.DimAsyncFunction;
import com.bean.OrderWide;
import com.bean.PaymentWide;
import com.bean.ProductStats;
import com.common.GlobalConfig;
import com.common.GmallConstant;
import com.utils.ClickHouseUtil;
import com.utils.DateTimeUtil;
import com.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.time.Duration;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;


/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2022/1/9 22:46
 */

//TODO: 数据流:
// 1. web/app -> Nginx -> Springboot  -> Kafka(ods)  ->  FlinkApp  -> Kafka(dwd) -> FlinkApp -> ClickHouse
// 2. web/app -> Nginx -> Springboot  -> Mysql  ->  FlinkApp  -> Kafka(ods)  -> FlinkApp -> Kafka/Phoenix(DWM) -> FlinkApp -> ClickHouse
//TODO: 程序 mockLog -> Nginx -> Logger.sh -> Kafka(ZK)/Phoenix(HDFS/HBase/ZK) -> Redis -> ClickHouse
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO:1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境，与Kafka分区数保持一致

        //TODO:Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
        /*//开启CK并指定状态后端为FS menory fs rocksdb
        env.setStateBackend(new FsStateBackend("hdfs://192.168.88.109:9820/gmall-flink/ck"));
        //开启 Checkpoint,每隔 5 秒钟做一次 CK
        env.enableCheckpointing(5000L);
        //指定 CK 的一致性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        //设置任务关闭的时候保留最后一次 CK 数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //指定从 CK 自动重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000L));
        //设置访问 HDFS 的用户名
        System.setProperty("HADOOP_USER_NAME", "root");*/

        //TODO:2.读取Kafka 7个主题的数据  创建数据流
        String groupId = "product_stats_app" + GlobalConfig.NUMBER;
        String pageViewSourceTopic = "dwd_page_log";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String cartInfoSourceTopic = "dwd_cart_info";
        String favorInfoSourceTopic = "dwd_favor_info";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";

        DataStreamSource<String> pvDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(pageViewSourceTopic, groupId));

        DataStreamSource<String> favoDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(favorInfoSourceTopic, groupId));

        DataStreamSource<String> cartDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(cartInfoSourceTopic, groupId));

        DataStreamSource<String> orderDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(orderWideSourceTopic, groupId));

        DataStreamSource<String> payDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(paymentWideSourceTopic, groupId));

        DataStreamSource<String> refundDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(refundInfoSourceTopic, groupId));

        DataStreamSource<String> commentDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(commentInfoSourceTopic, groupId));

        //TODO:3.将7个流 统一数据格式
        SingleOutputStreamOperator<ProductStats> productStatsWithClickAndDisplayDS =
                pvDS.flatMap(new FlatMapFunction<String, ProductStats>() {
                    @Override
                    public void flatMap(String value, Collector<ProductStats> out) throws Exception {
                        //将数据转换为JSON对象
                        JSONObject jsonObject = JSON.parseObject(value);

                        //取出page信息
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");

                        String page_id = page.getString("page_id");
                        if ("good_detail".equals(page_id) && "sku_id".equals(page.getString("item_type"))) {
                            out.collect(ProductStats.builder()
                                    .sku_id(page.getLong("item"))
                                    .click_ct(1L)
                                    .ts(ts)
                                    .build());
                        }

                        //尝试取出曝光数据
                        JSONObject displays = jsonObject.getJSONObject("displays");
                        if (displays != null && displays.size() > 0) {
                            for (int i = 0; i < displays.size(); i++) {

                                //取出单条曝光数据
                                if ("sku_id".equals(displays.getString("item_type"))) {
                                    out.collect(ProductStats.builder()
                                            .sku_id(displays.getLong("item"))
                                            .display_ct(1L)
                                            .ts(ts)
                                            .build());
                                }

                            }
                        }
                    }
                });

        SingleOutputStreamOperator<ProductStats> productStatsWithFavorDS =
                favoDS.map(line -> {
                    JSONObject jsonObject = JSON.parseObject(line);
                    return ProductStats.builder()
                            .sku_id(jsonObject.getLong("sku_id"))
                            .favor_ct(1L)
                            .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                            .build();
                });


        SingleOutputStreamOperator<ProductStats> productStatsWithCartDS =
                cartDS.map(line -> {
                    JSONObject jsonObject = JSON.parseObject(line);
                    return ProductStats.builder()
                            .sku_id(jsonObject.getLong("sku_id"))
                            .favor_ct(1L)
                            .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                            .build();
                });


        SingleOutputStreamOperator<ProductStats> productStatsWithOrderDS =
                orderDS.map(line -> {
                    OrderWide orderWide = JSON.parseObject(line, OrderWide.class);
                    HashSet<Long> orderIds = new HashSet<>();
                    orderIds.add(orderWide.getOrder_id());

                    return ProductStats.builder()
                            .sku_id(orderWide.getSku_id())
                            .order_sku_num(orderWide.getSku_num())
                            .order_amount(orderWide.getOrder_price())
                            .orderIdSet(orderIds)
                            .ts(DateTimeUtil.toTs(orderWide.getCreate_time()))
                            .build();
                });

        SingleOutputStreamOperator<ProductStats> productStatsWithPaymentDS = payDS.map(line -> {
            PaymentWide paymentWide = JSON.parseObject(line, PaymentWide.class);

            HashSet<Long> orderIds = new HashSet<>();
            orderIds.add(paymentWide.getOrder_id());

            return ProductStats.builder()
                    .sku_id(paymentWide.getSku_id())
                    .payment_amount(paymentWide.getOrder_price())
                    .paidOrderIdSet(orderIds)
                    .ts(DateTimeUtil.toTs(paymentWide.getPayment_create_time()))
                    .build();

        });

        SingleOutputStreamOperator<ProductStats> productStatsWithRefundDS =
                refundDS.map(line -> {
                    JSONObject jsonObject = JSON.parseObject(line);

                    HashSet<Long> orderIds = new HashSet<>();
                    orderIds.add(jsonObject.getLong("order_id"));

                    return ProductStats.builder()
                            .sku_id(jsonObject.getLong("sku_id"))
                            .refund_amount(jsonObject.getBigDecimal("refund_amount"))
                            .refundOrderIdSet(orderIds)
                            .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                            .build();
                });

        SingleOutputStreamOperator<ProductStats> productStatsWithCommentDS =
                commentDS.map(line -> {
                    JSONObject jsonObject = JSON.parseObject(line);

                    HashSet<Long> orderIds = new HashSet<>();
                    orderIds.add(jsonObject.getLong("order_id"));
                    String appraise = jsonObject.getString("appraise");
                    long goodCt = 0L;
                    if (GmallConstant.APPRAISE_GOOD.equals(appraise)) {
                        goodCt = 1L;
                    }

                    return ProductStats.builder()
                            .sku_id(jsonObject.getLong("sku_id"))
                            .comment_ct(1L)
                            .good_comment_ct(goodCt)
                            .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                            .build();
                });
        //TODO:4.Union 7个流

        DataStream<ProductStats> unionDS =
                productStatsWithClickAndDisplayDS.union(
                        productStatsWithFavorDS,
                        productStatsWithCartDS,
                        productStatsWithOrderDS,
                        productStatsWithPaymentDS,
                        productStatsWithRefundDS,
                        productStatsWithCommentDS
                );


        //TODO:5.提取时间戳生成WaterMark
        SingleOutputStreamOperator<ProductStats> productStatsWithWMDS =
                unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));


        //TODO:6.分组、开窗、聚合  按照sku_id 分组，10s 的滚动窗口，结合增量聚合（累加值）和全量聚合（提取窗口信息）

        SingleOutputStreamOperator<ProductStats> reduceDS =
                productStatsWithWMDS.keyBy(ProductStats::getSku_id)
                        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                        .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                                stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                                stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                                stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                                stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());

                                stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                                stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                                //stats1.setOrder_ct(stats1.getOrderIdSet().size() + 0L);
                                stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());

                                stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));
                                stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                                //stats1.setRefund_order_ct(stats1.getRefundOrderIdSet().size() + 0L);

                                stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));
                                stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                                //stats1.setPaid_order_ct(stats1.getPaidOrderIdSet().size() + 0L);
                                stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                                stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());
                                return stats1;
                            }
                        }, new WindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void apply(Long aLong, TimeWindow timeWindow, Iterable<ProductStats> iterable, Collector<ProductStats> collector) throws Exception {

                                //取出数据
                                ProductStats productStats = iterable.iterator().next();

                                //设置窗口时间
                                productStats.setStt(DateTimeUtil.toYMDhms(new Date(timeWindow.getStart())));
                                productStats.setEdt(DateTimeUtil.toYMDhms(new Date(timeWindow.getEnd())));

                                //设置订单数量
                                productStats.setOrder_ct(productStats.getOrderIdSet().size() + 0L);
                                productStats.setPaid_order_ct(productStats.getPaidOrderIdSet().size() + 0L);
                                productStats.setRefund_order_ct(productStats.getRefundOrderIdSet().size() + 0L);


                                //将数据写出
                                collector.collect(productStats);
                            }
                        });


        //TODO:7.关联维度信息
        //7.1关联SKU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSKUDS =
                AsyncDataStream.unorderedWait(reduceDS,
                        new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {

                            @Override
                            public void join(ProductStats productStats, JSONObject dimInfo) throws ParseException {


                                productStats.setSku_name(dimInfo.getString("SKU_NAME"));
                                productStats.setSku_price(dimInfo.getBigDecimal("PRICE"));
                                productStats.setSpu_id(dimInfo.getLong("SPU_ID"));
                                productStats.setTm_id(dimInfo.getLong("TM_ID"));
                                productStats.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));
                            }

                            @Override
                            public String getKey(ProductStats productStats) {
                                return productStats.getSku_id().toString();
                            }
                        }, 60, TimeUnit.SECONDS);


        //7.2关联SPU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSpuDS =
                AsyncDataStream.unorderedWait(productStatsWithSKUDS,
                        new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws ParseException {
                                productStats.setSpu_name(jsonObject.getString("SPU_NAME"));
                            }

                            @Override
                            public String getKey(ProductStats productStats) {
                                return productStats.getSpu_id().toString();
                            }
                        }, 60, TimeUnit.SECONDS);


        //7.3关联TM维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategory3DS =
                AsyncDataStream.unorderedWait(productStatsWithSpuDS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws ParseException {

                                productStats.setCategory3_name(jsonObject.getString("NAME"));
                            }

                            @Override
                            public String getKey(ProductStats productStats) {
                                return productStats.getCategory3_id().toString();
                            }
                        }, 60, TimeUnit.SECONDS);


        //7.4关联Category维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTmDS =
                AsyncDataStream.unorderedWait(productStatsWithCategory3DS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws ParseException {

                                productStats.setTm_name(jsonObject.getString("TM_NAME"));
                            }

                            @Override
                            public String getKey(ProductStats productStats) {
                                return productStats.getTm_id().toString();
                            }
                        }, 60, TimeUnit.SECONDS);


        //TODO:9.将数据写入ClickHouse
        System.out.println("任务开始>>>>>>>>>>>>>>>>>>");
        productStatsWithTmDS
                .addSink(ClickHouseUtil
                        .<ProductStats>getSink("insert into product_stats_2021 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));


        //TODO:10.启动任务
        env.execute("ProductStatsApp");

    }
}
